package io.kiki.sba.registry.server.data_store.multi.cluster.dataserver.handler;

import io.kiki.sba.registry.common.model.slot.DataSlotDiffDigestRequest;
import io.kiki.sba.registry.common.model.slot.SlotAccess;
import io.kiki.sba.registry.server.data_store.multi.cluster.executor.MultiClusterExecutorManager;
import io.kiki.sba.registry.server.data_store.remoting.dataserver.handler.BaseSlotDiffDigestRequestHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.concurrent.Executor;


public class MultiClusterSlotDiffDigestRequestHandler extends BaseSlotDiffDigestRequestHandler {

    private static final Logger logger = LoggerFactory.getLogger(MultiClusterSlotDiffDigestRequestHandler.class);

    @Autowired
    private MultiClusterExecutorManager multiClusterExecutorManager;

    public MultiClusterSlotDiffDigestRequestHandler() {
        super();
    }


    @Override
    public Executor getExecutor() {
        return multiClusterExecutorManager.getRemoteSlotSyncProcessorExecutor();
    }

    @Override
    protected boolean preCheck(DataSlotDiffDigestRequest request) {
        return doCheck(request);
    }

    private boolean doCheck(DataSlotDiffDigestRequest request) {
        if (!slotManager.isLeader(serverConfig.getDataCenter(), request.getSlotId())) {
            logger.warn("sync slot request from {}, not leader of {}", request.getLocalDataCenter(), request.getSlotId());
            return false;
        }

        SlotAccess slotAccess = slotManager.checkSlotAccess(serverConfig.getDataCenter(), request.getSlotId(), request.getSlotsEpoch(), request.getSlotLeaderEpoch());

        if (!slotAccess.isAccept()) {
            logger.warn("sync slot request from {}, slotId={}, slotAccess={}, request={}", request.getLocalDataCenter(), request.getSlotId(), slotAccess, request);
        }

        return slotAccess.isAccept();
    }

    @Override
    protected boolean postCheck(DataSlotDiffDigestRequest request) {
        return doCheck(request);
    }
}
